package com.sg.java.apps;

import com.google.gson.Gson;
import com.google.gson.JsonArray;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.sg.java.PropertiesUtil;
import com.sg.java.ResourcePath;
import com.sg.java.util.HBaseUtils;
import com.sg.java.util.HdfsUtils;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.PrintWriter;
import java.util.Arrays;
import java.util.List;

public class WriteWGToTxt {

    private static final Logger log = LoggerFactory.getLogger(HBaseToHdfs.class);

    public static byte[] info_column         = Bytes.toBytes("info");
    public static byte[] time_qualifier      = Bytes.toBytes("time");
    public static byte[] mRId_qualifier      = Bytes.toBytes("mRId");
    public static byte[] areaType_qualifier  = Bytes.toBytes("areaType");
    public static byte[] equipType_qualifier = Bytes.toBytes("equipType");
    public static byte[] analog_qualifier    = Bytes.toBytes("Analog");
    public static Gson   gson                = new Gson();

//    public static List<String> types = Lists.newArrayList(
//            "PhW_phsA",
//            "PhW_phsB",
//            "PhW_phsC",
//            "PhVAr_phsA",
//            "PhVAr_phsB",
//            "PhVAr_phsC",
//            "TotW",
//            "TotVAr"
//                                                         );

    //java -jar WriteWGToTxt dms_decloud_mlv_measuration_20221110 /dms_decloud_mlv_measuration_20221110
    public static void main(String[] args) throws Exception {
        String         name     = args[0];
        String         hdfsPath = args[1].trim();
        String         typesStr = args[2];
        final String[] types    = typesStr.split(",");

        List<String>   typeList = Arrays.asList(types);
        int            batch    = 500000;
        if (!hdfsPath.startsWith(HdfsUtils.hdfsPathPrefix)) {
            hdfsPath = HdfsUtils.hdfsPathPrefix + hdfsPath;
        }

        Connection conn      = HBaseUtils.getHBaseConn(PropertiesUtil.createPropertiesFromResource(ResourcePath.hbase_properties));
        TableName  tableName = TableName.valueOf(HBaseUtils.withNamespace(name));
        Table      table     = conn.getTable(tableName);

        Path               path = new Path(hdfsPath);
        FSDataOutputStream os   = HdfsUtils.fs.create(path, true);
        PrintWriter        pw   = new PrintWriter(os);

        long   total   = 0;
        long   count;
        byte[] nextRow = null;

        do {
            count = 0;
            Scan scan = new Scan();
            scan.setLimit(batch);
            if (nextRow != null) {
                scan.withStartRow(nextRow, false);
            }
            scan.setCaching(1000);
            ResultScanner scanner = table.getScanner(scan);
            for (Result result : scanner) {
                count++;
                nextRow = result.getRow();
                JsonObject jsonObject = new JsonObject();
                jsonObject.addProperty("mRId", Bytes.toString(result.getValue(info_column, mRId_qualifier)));
                jsonObject.addProperty("time", Bytes.toString(result.getValue(info_column, time_qualifier)));
                jsonObject.addProperty("areaType", Bytes.toString(result.getValue(info_column, areaType_qualifier)));
                jsonObject.addProperty("equipType", Bytes.toString(result.getValue(info_column, equipType_qualifier)));
                if (!jsonObject.get("equipType").getAsString().equals("PD_5006")) {
                    continue;
                }
                final String    analog     = Bytes.toString(result.getValue(info_column, analog_qualifier));
                final JsonArray analogList = gson.fromJson(analog, JsonArray.class);
                for (JsonElement jsonElement : analogList) {
                    final JsonObject jo              = jsonElement.getAsJsonObject();
                    final String     measurementType = jo.get("measurementType").getAsString();
                    if (typeList.contains(measurementType)) {
                        String value = jo.get("value").getAsString();
                        jsonObject.addProperty(measurementType, value);
                    }
                }
                pw.println(jsonObject);
            }
            total += count;
            log.info("已写入:{}", total);
        }
        while (count == batch);

        pw.flush();
        pw.close();
        os.close();
        table.close();
        conn.close();
    }

}
